package com.thp.bigdata.eachFan;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class EachFan {
	static class EachFanMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
		Text k = new Text();
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
				String line = value.toString();
				String user = line.split(":")[0];
				String[] friends = line.split(":")[1].split(",");
				for(int i = 0; i < friends.length; i++) {
					String friend = friends[i];
					/**
					 * 下面这个过程就是关键的过程，
					 * 第一行数据 ： A:B,C,D,F,E,O
					 * 第二行数据  ：B:A,C,E,K
					 * A里面有B，B里面有A
					 * 现在我们需要获取这两个
					 * 那么我们在map的时候， user 跟 friend 进行组合往reduce输出的时候，
					 * 我们组合的时候需要进行排序，  
					 * 第一行数据组合  A-B
					 * 第二行如果没有排序 那么就是  B-A
					 * 但是我们在reduce进行处理时，需要的是相同的key
					 * 那么我们应该让 A-B  与  B-A 变成同一个key往reduce输出
					 * 这样reduce才能根据相同的key进行数量的统计
					 * 当数量为2时，就代表，A-B出现了两次，那么这一对手机互相关注的。
					 */
					if(user.compareTo(friend) < 0) {
						k.set(user + "-" + friend);
						context.write(k, NullWritable.get());
					} else {
						k.set(friend + "-" + user);
						context.write(k, NullWritable.get());
					}
				}
		}
	}	
	
	static class EachFanReducer extends Reducer<Text, NullWritable, Text, NullWritable> {
		@Override
		protected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
			int count = 0;
			for(NullWritable nw : values) {
				count++;
			}
			/**
			 * 严格来说，这个count要么为1 要么为 2，当为2的时候，就说明是互粉
			 */
			if(count == 2) {
				context.write(key, NullWritable.get());
			}
		}
	}
	
	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);
		job.setJarByClass(EachFan.class);
		
		job.setMapperClass(EachFanMapper.class);
		job.setReducerClass(EachFanReducer.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(NullWritable.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(NullWritable.class);
		
		FileInputFormat.setInputPaths(job, new Path("F:/commonFriends/input"));
		FileOutputFormat.setOutputPath(job, new Path("F:/commonFriends/eachFan_out"));
		
		System.exit(job.waitForCompletion(true) ? 0 : 1);
		
	}
	
}
